package DianShang_2024.ds_01.clean

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{current_timestamp, date_format, lit, to_timestamp}

object clean03 {
  def main(args: Array[String]): Unit = {
        /*
            抽取ods库base_province表中昨天的分区（子任务一生成的分区）数据，并结合dim_province最新分区现有的数据，根据id合并数据到dwd库中dim_province
            的分区表（合并是指对dwd层数据进行插入或修改，需修改的数据以id为合并字段，根据create_time排序取最新的一条），分区字段为etl_date且值与ods库的相
            对应表该值相等，并添加dwd_insert_user、dwd_insert_time、dwd_modify_user、dwd_modify_time四列,其中dwd_insert_user、
            dwd_modify_user均填写“user1”。若该条数据第一次进入数仓dwd层则dwd_insert_time、dwd_modify_time均填写当前操作时间，并进行数据类型转换。
            若该数据在进入dwd层时发生了合并修改，则dwd_insert_time时间不变，dwd_modify_time存当前操作时间，其余列存最新的值。使用hive cli在
            表dwd.dim_province最新分区中，查询该分区中数据的条数
         */
    //  准备sparksql的环境
    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("数据清洗第三题")
      .enableHiveSupport()
      .getOrCreate()

    //  在插入数据之前，数据只有1-21行，id不相同没有重复的
    spark.sql("use dwd")
    spark.sql("select * from dwd.dim_province").show

    //  首先对ods.base_province2表进行添加字段操作
    spark.sql("select * from ods.base_province2 where etl_date='20231017'")
      .drop("etl_date")
      .withColumn("dwd_insert_user",lit("user1"))
      .withColumn(
        "dwd_insert_time",
        lit(to_timestamp(date_format(current_timestamp(),"yyyy-MM-dd HH:mm:ss"),"yyyy-MM-dd HH:mm:ss"))
      )
      .withColumn("dwd_modify_user",lit("user1"))
      .withColumn(
        "dwd_modify_time",
        lit(to_timestamp(date_format(current_timestamp(),"yyyy-MM-dd HH:mm:ss"),"yyyy-MM-dd HH:mm:ss"))
      )
      .withColumn("etl_date",lit("20231017"))
      .createOrReplaceTempView("base_province_temp")

    //  t1:将dwd.dim_provice和ods.base_province2的临时表进行合并，并且在外层进行窗口排列
    spark.sql(
      """
        |select
        |id,
        |name,
        |region_id,
        |area_code,
        |iso_code,
        |create_time,
        |row_number() over(partition by id order by create_time desc) as row,
        |dwd_insert_user,
        |min(dwd_insert_time) over(partition by id) as dwd_insert_time,
        |dwd_modify_user,
        |dwd_modify_time
        |from(
        |select * from dwd.dim_province
        |union all
        |select * from base_province_temp
        |) as merge_table
        |""".stripMargin).createOrReplaceTempView("t1")

    //  根据t1合并后结果拿取窗口里面第一行的数据插入dwd.dim_province
    spark.sql(
      """
        |insert into table dwd.dim_province
        |partition(etl_date='20231017')
        |select
        |id,
        |name,
        |region_id,
        |area_code,
        |iso_code,
        |create_time,
        |dwd_insert_user,
        |dwd_insert_time,
        |dwd_modify_user,
        |dwd_modify_time
        |from t1
        |where row=1
        |""".stripMargin)





    //  关闭sparksql的环境
      spark.close()
  }

}
